package com.lazar.module1.subject5

import java.util.UUID

import akka.actor.{Actor, ActorRef, ActorSelection, ActorSystem, Props}
import com.typesafe.config.{Config, ConfigFactory}

/*
1 向master发送注册信息
2 接收master的回复消息
 */
class SparkWorker extends Actor {
  var masterSelection: ActorSelection = _

  override def preStart(): Unit = {
    //根据master的地址获取到masteractor的引用对象
    //通过actorselection获取
    masterSelection = context.actorSelection(s"akka.tcp://SparkMaster@127.0.0.1:10000/user/SparkMaster01")
  }


  override def receive: Receive = {
    //worker启动后向master发送注册请求
    case "start" => {
      println("worker 已经启动。。。")
      masterSelection ! RegisterWorkerInfo(UUID.randomUUID().toString, 2, 2)
    }
    //服务端注册成功的回复
    case RegisteredWorkerInfo=>{
      println("收到master注册成功的回复消息。。。")

      //开启一个定时发送消息到master的任务
      //1.schedule 创建一个定时器
      // 2.0 millis, 延时多久才执行, 0 表示不延时，立即执行
      // 3. 3000 millis 表示每隔多长时间执行 3 秒
      // 4. self 给自己发送 消息
      // 5. SendHeartBeat 消息
      import scala.concurrent.duration._
      val actorSystem: ActorSystem = context.system
      import actorSystem.dispatcher

      context.system.scheduler.schedule(0 millis, 3000 millis, self, SendHeartBeat)
    }
  }
}

object SparkWorker extends App {
  //跨节点(网络)与worker通信
  val masterHost = "127.0.0.1"
  val masterPort = 10000

  val workerHost = "127.0.0.1"
  val workerPort = 10001

  val config: Config = ConfigFactory.parseString(
    s"""
akka.actor.provider="akka.remote.RemoteActorRefProvider"
akka.remote.netty.tcp.hostname=$workerHost
akka.remote.netty.tcp.port=$workerPort
""".stripMargin
  )

  //创建actorsystem
  val sparkWorkerActorSystem = ActorSystem("SparkWorker", config)
  //注册actor
  val sparkWorker0Ref: ActorRef = sparkWorkerActorSystem.actorOf(Props[SparkWorker], "SparkWorker01")
  //启动masteractor
  sparkWorker0Ref ! "start"


}